package com.atguigu.flink.streamapi.transform;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/21
 *      KeyedStream.process(KeyedProcessFunction x)
 *
 *      SingleOutputStreamOperator<WaterSensor>.process(ProcessFunction x )
 *
 */
public class Demo11_ProcessKeyBy
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .keyBy(w -> "a")
            .process(new MyProcess())
            .print();


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }

    // KeyedProcessFunction<Key的类型, 输入的类型, 输出的类型>
    public static class MyProcess extends KeyedProcessFunction<String, WaterSensor, String>
    {
        int sumVc = 0;
        /*
             WaterSensor value: 输入的元素
              Context ctx： 编程环境，从里面获取一些信息
              Collector<String> out： 输出
         */
        @Override
        public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
            System.out.println("当前的key："+ctx.getCurrentKey());
            sumVc += value.getVc();
            out.collect("当前累积水位:"+sumVc);
        }
    }
}
